package com.to8to.scene.deamon;

import static com.to8to.scene.server.Commons.mongo;
import it.sauronsoftware.cron4j.Scheduler;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Maps;
import com.mongodb.AggregationOutput;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.to8to.commons.mq.IMessageHandler;
import com.to8to.commons.utils.ThriftUtil;
import com.to8to.kitt.esb.DefaultDeamon;
import com.to8to.kitt.esb.NodeChangeEvent;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.scene.server.Commons;
import com.to8to.scene.server.Tables;
import com.to8to.thrift.diary.consts.constConstants;
import com.to8to.thrift.diary.module.TComment;
import com.to8to.thrift.diary.module.TUserScore;
import com.to8to.thrift.diary.module.TUserScore._Fields;
import com.to8to.thrift.diary.service.SceneDiaryService;

/**
 * UserComputer.java
 * 作 者:atom.liu
 * 版 本:v2.0
 * 日期:2014年11月8日
 * 描 述:用户活跃积分（达人榜）计算程序。发布一篇日记20分，回复一篇日记5分，现场被收藏1分，日记被回复1分。
 */
public class UserComputer extends DefaultDeamon implements IMessageHandler
{
    public static Logger logger = LoggerFactory.getLogger(UserComputer.class);

    public static String LOGBACKConfPath = "crm/common/logback.xml";

    /**
     * 用户分数缓存Map。key：userId， value：TUserScore对象
     */
    private Map<String, TUserScore> userMap = Maps.newHashMap();

    /**
     * 不包含的字段集合。
     */
    private List<String> exclude_fields = new ArrayList<String>();

    private static Scheduler scheduler = new Scheduler();

    private static volatile boolean isFirst = false;

    /**
     * 执行计算。每天凌晨三点计算出前三十天用户的活跃积分。
     */
    public void start()
    {
        if (!isFirst)
        {
            scheduler.schedule(Commons.user_compute_schedule_pattern, new Runnable()
            {
                @Override
                public void run()
                {
                    long t1 = System.currentTimeMillis();
                    user_score();
                    flush2db();
                    long t2 = System.currentTimeMillis();
                    logger.info("finish compute user score, use time: " + (t2 - t1) + " ms");
                }
            });

            isFirst = true;
        }
        scheduler.start();
    }

    public void stop()
    {
        if (scheduler.isStarted())
            scheduler.stop();
    }

    /**
     * 计算用户各种行为个数
     */
    private void user_score()
    {
        int start_time_sec = (int) (new Date().getTime() / 1000) - 3600 * 24 * 30;

        for (_Fields field : TUserScore._Fields.values())
            exclude_fields.add(field.getFieldName());

        exclude_fields.remove(user_diary_num(start_time_sec));

        exclude_fields.remove(user_comment_num(start_time_sec));

        exclude_fields.remove(user_becomment_num(start_time_sec));

        exclude_fields.remove(user_scene_coll_num(start_time_sec));

        exclude_fields.remove(TUserScore._Fields.SCORE.getFieldName());

        logger.info("compute user_score[finish]");
    }

    /**
     * 计算分数，刷新到数据库
     */
    private void flush2db()
    {
        logger.info("compute user_score and flush2db");

        for (TUserScore user : userMap.values())
        {
            int score =
                user.getDiary_num() * 20 + user.getComment_num() * 5 + user.getBecomment_num() + user.getScene_coll_num() + user.getDiary_coll_num();
            user.setScore(score);
            DBObject dbobj = ThriftUtil.thriftObject2DBObject(user);
            for (String field : exclude_fields)
                dbobj.removeField(field);

            mongo.update(Tables.UserScore, new BasicDBObject("_id", user.get_id()), new BasicDBObject("$set", dbobj), true, false);
            logger.debug("user " + user.get_id() + "update for :" + dbobj);
        }

        for (String uid : mongo.ids(Tables.UserScore))
        {
            if (!userMap.containsKey(uid))
            {
                TUserScore user = new TUserScore();
                user.set_id(uid);
                DBObject dbobj = ThriftUtil.thriftObject2DBObject(user);
                for (String field : exclude_fields)
                    dbobj.removeField(field);
                mongo.update(Tables.UserScore, new BasicDBObject("_id", user.get_id()), new BasicDBObject("$set", dbobj), true, false);
                logger.debug("user " + user.get_id() + "update for :" + dbobj);
            }
        }
        userMap = null;
        userMap = Maps.newHashMap();

        logger.info("flush2db [finish]");
    }

    /**
     * 计算用户最近30天发表审核通过的日记数 
     */
    @SuppressWarnings("deprecation")
    private String user_diary_num(int start_time_sec)
    {
        logger.info("compute user_diary_num");

        BasicDBObject group = new BasicDBObject("$group", new BasicDBObject("_id", "$owner_id").append("num", new BasicDBObject("$sum", 1)));

        BasicDBObject query = new BasicDBObject("status", constConstants.ENTITY_STATUS_NORM);

        if (start_time_sec > 0)
            query.put("ctime", new BasicDBObject("$gte", start_time_sec));

        BasicDBObject match = new BasicDBObject("$match", query);

        AggregationOutput output = mongo.getDb().getCollection(Tables.Diary).aggregate(match, group);

        for (DBObject obj : output.results())
        {
            BasicDBObject dbobj = (BasicDBObject) obj;
            String user_id = dbobj.getString("_id");
            if (StringUtils.isEmpty(user_id))
                continue;
            int diary_num = dbobj.getInt("num", 0);
            TUserScore user = userMap.get(user_id);
            if (user == null)
            {
                user = new TUserScore();
                user.set_id(user_id);
                userMap.put(user_id, user);
            }
            user.setDiary_num(diary_num);
        }
        logger.info("compute user_diary_num[finish]");
        return TUserScore._Fields.DIARY_NUM.getFieldName();
    }

    /**
     * 用户日记被回复数
     */
    @SuppressWarnings("deprecation")
    private String user_becomment_num(int start_time_sec)
    {
        logger.info("compute user_becomment_num");
        BasicDBObject group = new BasicDBObject("$group", new BasicDBObject("_id", "$owner_id").append("num", new BasicDBObject("$sum", 1)));
        BasicDBObject query = new BasicDBObject("status", constConstants.ENTITY_STATUS_NORM);
        query.put(TComment._Fields.TGT_TYPE.getFieldName(), constConstants.TARGET_TYPE_DIARY);
        if (start_time_sec > 0)
            query.put("ctime", new BasicDBObject("$gte", start_time_sec));
        BasicDBObject match = new BasicDBObject("$match", query);
        AggregationOutput output = mongo.getDb().getCollection(Tables.Comment).aggregate(match, group);

        Set<String> idset = new HashSet<String>();
        for (DBObject obj : output.results())
        {
            BasicDBObject dbobj = (BasicDBObject) obj;
            String user_id = dbobj.getString("_id");

            if (StringUtils.isEmpty(user_id))
                continue;

            idset.add(user_id);
            int becomment_num = dbobj.getInt("num", 0);

            TUserScore user = userMap.get(user_id);
            if (user == null)
            {
                user = new TUserScore();
                user.set_id(user_id);
                userMap.put(user_id, user);
            }
            user.setBecomment_num(becomment_num);
        }

        logger.info("compute user_becomment_num[finish]");

        return TUserScore._Fields.BECOMMENT_NUM.getFieldName();
    }

    /**
     * 用户评论数
     */
    @SuppressWarnings("deprecation")
    private String user_comment_num(int start_time_sec)
    {
        logger.info("compute user_comment_num");

        BasicDBObject group = new BasicDBObject("$group", new BasicDBObject("_id", "$user_id").append("num", new BasicDBObject("$sum", 1)));

        BasicDBObject query = new BasicDBObject("status", constConstants.ENTITY_STATUS_NORM);

        query.put("tgt_type", constConstants.TARGET_TYPE_DIARY);

        if (start_time_sec > 0)
            query.put("ctime", new BasicDBObject("$gte", start_time_sec));

        BasicDBObject match = new BasicDBObject("$match", query);

        AggregationOutput output = mongo.getDb().getCollection(Tables.Comment).aggregate(match, group);

        Set<String> idset = new HashSet<String>();

        for (DBObject obj : output.results())
        {
            BasicDBObject dbobj = (BasicDBObject) obj;
            String user_id = dbobj.getString("_id");

            if (StringUtils.isEmpty(user_id))
                continue;

            idset.add(user_id);
            int comment_num = dbobj.getInt("num", 0);

            TUserScore user = userMap.get(user_id);

            if (user == null)
            {
                user = new TUserScore();
                user.set_id(user_id);
                userMap.put(user_id, user);
            }

            user.setComment_num(comment_num);
        }

        logger.info("compute user_comment_num[finish]");

        return TUserScore._Fields.COMMENT_NUM.getFieldName();
    }

    /**
     * 计算30天内用户现场的收藏数 
     */
    @SuppressWarnings("deprecation")
    private String user_scene_coll_num(int start_time_sec)
    {
        logger.info("compute user_scene_coll_num");

        BasicDBObject group = new BasicDBObject("$group", new BasicDBObject("_id", "$user_id").append("num", new BasicDBObject("$sum", 1)));

        BasicDBObject query = new BasicDBObject();

        if (start_time_sec > 0)
            query.put("ctime", new BasicDBObject("$gte", start_time_sec));

        query.put("tgt_type", constConstants.TARGET_TYPE_SCENE);
        query.put("bhv_type", constConstants.BEHAVIOR_TYPE_COLL);
        query.put("status", 1);

        BasicDBObject match = new BasicDBObject("$match", query);

        AggregationOutput output = mongo.getDb().getCollection(Tables.Behavior).aggregate(match, group);

        Set<String> idset = new HashSet<String>();

        for (DBObject obj : output.results())
        {
            BasicDBObject dbobj = (BasicDBObject) obj;
            String user_id = dbobj.getString("_id");

            if (StringUtils.isEmpty(user_id))
                continue;

            idset.add(user_id);
            int scene_coll_num = dbobj.getInt("num", 0);

            TUserScore user = userMap.get(user_id);

            if (user == null)
            {
                user = new TUserScore();
                user.set_id(user_id);
                userMap.put(user_id, user);
            }

            user.setScene_coll_num(scene_coll_num);
        }
        logger.info("compute user_scene_coll_num [finish]");

        return TUserScore._Fields.SCENE_COLL_NUM.getFieldName();
    }

    @Override
    public boolean handle(Delivery arg0) throws Exception
    {
        return false;
    }

    @Override
    public void onPrivateConfigChange(Map<String, NodeChangeEvent> config)
    {
        String[] keys = { "common.properties" };

        for (String file_name : keys)
        {
            NodeChangeEvent event = config.get(file_name);

            if (!event.isChange())
                continue;

            byte[] data = event.ndata;

            switch (file_name)
            {
            case "common.properties":
            {
                Commons.init(data);
                break;
            }
            }
        }

        super.onPrivateConfigChange(config);
    }

    @Override
    public void onCommonConfigChange(NodeChangeEvent e)
    {
        if (e.path.contains(LOGBACKConfPath))
        {
            String logbackConfig = new String(e.ndata);
            String real = logbackConfig.replace("${java.service}", service).replace("${java.instance}", instance);
            LogUtils.resetConfigByData(real);
        }
    }

    public static void main(String[] args) throws Exception
    {
        String instance = "compute_1";

        if (args.length > 0)
            instance = args[0];

        UserComputer deamon = new UserComputer();
        deamon.setReportAddressToZK(true);
        deamon.setSyncConfigFromZK(true);
        deamon.setServiceInstance(SceneDiaryService.class.getName(), instance);
        deamon.addCommonConfigPath(LOGBACKConfPath);
        deamon.init();
        deamon.stop();
        deamon.start();
        new CountDownLatch(1).await();
    }
}
